package com.codeduck.kafka.kafka;

import com.codeduck.kafka.kafka.consumer.IMessageConsumer;
import com.codeduck.kafka.kafka.producer.IMessageProducer;
import com.codeduck.kafka.exception.MessageException;
import com.codeduck.kafka.pojo.TopicHandler;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.Properties;

/**
 * @author: <a href="mailto:codeduck@163.com">Codeduck</a>
 * @date: 2023/3/4
 */
@Component
@Scope(BeanDefinition.SCOPE_PROTOTYPE)
public class KafkaInvoker {

    @Resource
    private IMessageProducer producer;
    @Resource
    private IMessageConsumer consumer;

    /**
     * 轮询到各分区
     */
    public void sendMessage(String message, String topic) throws MessageException {
        producer.sendMessage(message, topic);
    }

    /**
     * 指定分区
     */
    public void sendMessage(String message, String topic, int partition) throws MessageException {
        producer.sendMessage(message, topic, partition);
    }

    /**
     * 订阅topic
     */
    public void subscribeTopic(Properties properties, TopicHandler topicHandler) throws MessageException {
        consumer.subscribe(properties, topicHandler);
    }

    public void assignTopic(Properties properties, List<TopicPartition> topicPartitions) {
        consumer.assign(properties, topicPartitions);
    }

}
